package com.niit.service

import java.io.{FileWriter, PrintWriter}
import java.text.SimpleDateFormat

import com.niit.bean.AdClickData
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.DStream

import scala.collection.mutable.ListBuffer

class TimeCountService  {


  def dataAnalysis(data:DStream[AdClickData]): Unit ={

    /*
      统计一分钟内的广告点击量，每10秒合并一次
       x轴表示时间
          31:00  31:10  31:20  31:30  31:40 ……………………32：00
        31:44 点击的数据 ————> 44 / 10 = 4  --> 40点击的数据  因为是统计的是10秒
        31:48 点击数据  ————> 48 / 10 = 4
        1s 等于 1000毫秒  1000 * 10 = 10000
        17564789412345 ==> 2024-05-23 17:39:41
        17564789412345 / 10000 * 10000 = 1756478941 * 10000 == 17564789410000 => 2024-05-23 17:30:00
     */
    //1.将点击广告的时间戳 进行转换  例如 44秒点击的数据 改为 40秒点击的数据 49点击的数据改为40数据
    val reduceDS: DStream[(Long, Int)] = data.map(line => {
      val ts = line.ts.toLong //  17564789412345
      val newts = ts / 10000 * 10000
      (newts, 1) //    17564789412345  1
    }).reduceByKeyAndWindow(
      (x: Int, y: Int) => { // x:10秒内旧的点击次数  y:10秒内新的点击次数
        x + y

      }, Seconds(60), //统计的是一分钟内的数据
      Seconds(10) //每次10秒计算一次
    )
    /*
     x：31:00 (31:00--31:09)     y:15
        31:10 (31:10--31:19)       10
        31:20  (31:20--31:29)      19
      */

    //格式化数据 并将数据显示到页面中

    reduceDS.foreachRDD(rdd=>{
      //新建列表，用来存储格式化后的数据
      val list = ListBuffer[String]()
      //将数据中的时间进行排序
      val datas: Array[(Long, Int)] = rdd.sortByKey(true).collect()
      //得到的数据，进行遍历 并根据模式匹配对数据进行格式化
      datas.foreach{
        case (time,count)=>{
          //建立时间格式化对象 时 : 分 ：秒
          val sdf  = new SimpleDateFormat("HH:mm:ss")
          val timeStr = sdf.format(new java.util.Date(time.toLong))// -> 18:20:10
          //组装数据 根据 { "xtime":"59:50", "yval":"88" }
          list.append(s""" {"xtime":"${timeStr}","yval":"${count}" }""")
        }
      }
      //输出文件---输出adclick.josn
      val out = new PrintWriter(new FileWriter("data/adclick.json"))
      out.println("[" + list.mkString(",") +"]")
      out.flush()
      out.close()

    })


  }

}
